package net.lmc.realtime.jtp.dwd.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.lmc.bwie.realtime.jtp.common.utils.DateTimeUtil;
import net.lmc.bwie.realtime.jtp.common.utils.DorisUtil;
import net.lmc.bwie.realtime.jtp.common.utils.KafkaUtil;
import net.lmc.realtime.jtp.dwd.function.LoadProvinceDimMapFunction;
import net.lmc.realtime.jtp.dwd.function.RequestDicDimAsyncFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class JtpOrderDetailDorisDwdJob {
    public static void main(String[] args) throws Exception {
        //执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //数据源
        DataStream<String> dataStream = KafkaUtil.consumerKafka(env, "topic_db");
        //dataStream.print();

        //数据转换
        DataStream<String> dwdStream = handle(dataStream);


        DorisUtil.saveToDoris(dwdStream,"jtp_mall_doris","dwd_order_detail");

        //触发执行
        env.execute("JtpOrderDetailDorisDwdJob");
    }

    private static DataStream<String> handle(DataStream<String> dataStream) {
        //数据流join
        DataStream<String> joinStream = joinProcess(dataStream);

        SingleOutputStreamOperator<String> provinceStream = joinStream.map(new LoadProvinceDimMapFunction());

        SingleOutputStreamOperator<String> dwdStream = AsyncDataStream.orderedWait(
                provinceStream,
                new RequestDicDimAsyncFunction(),
                10000,
                TimeUnit.MILLISECONDS,
                100
        );

        return dwdStream;
    }

    private static DataStream<String> joinProcess(DataStream<String> dataStream) {

        //订单明细数据流
        SingleOutputStreamOperator<String> detailStream = dataStream.filter(new RichFilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                return "order_detail".equals(jsonObject.getString("table_name"))
                        && "insert".equals(jsonObject.getString("operate_type"));
            }
        });

        SingleOutputStreamOperator<String> infoStream = dataStream.filter(new RichFilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                return "order_info".equals(jsonObject.getString("table_name"))
                        && "insert".equals(jsonObject.getString("operate_type"));
            }
        });

        SingleOutputStreamOperator<String> timeInfoStream = infoStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                            @Override
                            public long extractTimestamp(String element, long recordTimestamp) {
                                String eventTime = JSON.parseObject(element)
                                        .getJSONObject("operate_data").getString("create_time");
                                return DateTimeUtil.convertStringToLong(eventTime,"yyyy-MM-dd HH:mm:ss");
                            }
                        })
        );

        SingleOutputStreamOperator<String> timeDetailStream = detailStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                            @Override
                            public long extractTimestamp(String element, long recordTimestamp) {
                                String eventTime = JSON.parseObject(element)
                                        .getJSONObject("operate_data").getString("create_time");
                                return DateTimeUtil.convertStringToLong(eventTime, "yyyy-MM-dd HH:mm:ss");
                            }
                        })

        );

        SingleOutputStreamOperator<String> joinStream = timeInfoStream
                .keyBy(
                        element -> JSON.parseObject(element).getJSONObject("operate_data").getString("id")
                )
                .intervalJoin(
                        timeDetailStream.keyBy(
                                element -> JSON.parseObject(element).getJSONObject("operate_data").getString("order_id")
                        )
                )
                .between(Time.seconds(-1), Time.seconds(2))
                .process(new ProcessJoinFunction<String, String, String>() {
                    @Override
                    public void processElement(String info, String detail, ProcessJoinFunction<String, String, String>
                            .Context ctx, Collector<String> out) throws Exception {
                        JSONObject infoJsonObject = JSON.parseObject(info).getJSONObject("operate_data");
                        Integer userId = infoJsonObject.getInteger("user_id");
                        Integer provinceId = infoJsonObject.getInteger("province_id");

                        JSONObject detailJsonObject = JSON.parseObject(detail).getJSONObject("operate_data");
                        detailJsonObject.put("user_id", userId);
                        detailJsonObject.put("province_id", provinceId);

                        String createTime = detailJsonObject.getString("create_time");
                        String curDate = createTime.substring(0, 10);
                        detailJsonObject.put("cur_date", curDate);

                        String output = detailJsonObject.toJSONString();
                        out.collect(output);
                    }
                });

        return joinStream;
    }
}
